package com.gopay.common.mq.consumer;


import com.gopay.common.mq.configuration.MqConfig;
import com.gopay.common.mq.converter.JsonMsgConverter;
import com.gopay.common.mq.util.MqUtils;
import com.gopay.common.util.ClassesUtil;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.listener.DefaultMessageListenerContainer;

import javax.jms.Message;
import javax.jms.MessageListener;
import java.io.Serializable;

/**
 * 消息消费者基类
 * @param <T> 消息类型
 */
public abstract class AbstractMsgConsumer<T extends Serializable> extends DefaultMessageListenerContainer implements MessageListener, BeanFactoryAware{

    protected Logger logger = LoggerFactory.getLogger(getClass());

    private BeanFactory beanFactory ;

    private Class<T> msgClass = ClassesUtil.getSuperClassGenericType(getClass());

    @Autowired
    private MqConfig mqConfig;

    @Override
    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }

    private JsonMsgConverter<T> msgConverter = new JsonMsgConverter(msgClass);

    /**
     * 接收消息
     */
    @Override
    public void onMessage(Message message) {
        try {
            if(logger.isDebugEnabled()) {
                logger.info("Receive message :{}", message);
            }

            T msg = msgConverter.fromMessage(message);

            logger.info("Call doOnMessage !");
            doOnMessage(msg);
            logger.info("Call doOnMessage over !");

        } catch (Exception e) {
            logger.error("Something wrong when processing onMessage .", e);
            throw new RuntimeException(e.getMessage());
        }

        logger.info("Process onMessage over !");
    }



    public abstract void doOnMessage(T msg) throws Exception;

    @Override
    public void afterPropertiesSet() {
        ActiveMQConnectionFactory connectionFactory = beanFactory.getBean(ActiveMQConnectionFactory.class);
        if(connectionFactory == null){
            logger.error("Not Found ConnectionFactory in BeanFactory ." );
            throw new RuntimeException("Not Found ConnectionFactory in BeanFactory .");
        }
        setConnectionFactory(connectionFactory);
        setMessageListener(this);
        setSessionTransacted(true);

        toCompleteInit();

        processDestinationSuffix();

        super.afterPropertiesSet();
    }

    /**
     * 完成消费者初始化动作  比如：设置destination等。
     */
    public abstract void toCompleteInit();

    private void processDestinationSuffix(){
        MqUtils.appendSuffix(getDestination(), mqConfig.SUFFIX);
    }
}